package com.gy.flink.state;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class OperatorStateMain {

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        env.setStateBackend(new MemoryStateBackend());
//        env.setStateBackend(new FsStateBackend("./FsStateBackend"));
//        env.setStateBackend(new RocksDBStateBackend("./FsStateBackend"));

        //每隔时间自动执行一次checkpoint
        env.enableCheckpointing(30000);

//        //仅有一次语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //两次checkpoing最少间隔时间
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//        //超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(30000);
//        //checkpoing最大保留的批次
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(10);
//        //RETAIN_ON_CANCELLATION重启后恢复旧checkpoing的数据    DELETE_ON_CANCELLATION重启后删除旧数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);


        //不重启
//        env.setRestartStrategy(RestartStrategies.noRestart());
        //认识尝试重启3次 每次间隔10秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3,
                Time.of(10, TimeUnit.SECONDS)
        ));
//        //5分钟内最多重试3次，第3次失败旧不重试了  失败后间隔10秒重试
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3,
//                Time.of(5, TimeUnit.MINUTES),
//                Time.of(10, TimeUnit.SECONDS)
//        ));

        DataStreamSource<Tuple2<String, Integer>> dataStream = env.fromElements(
                Tuple2.of("spark", 2),
                Tuple2.of("hadoop", 7),
                Tuple2.of("hive", 4),
                Tuple2.of("flink", 12)
        );

        dataStream.addSink(new SinkFunctionOperatorState(2)).setParallelism(1);


        env.execute(OperatorStateMain.class.getSimpleName());
    }


    public static class SinkFunctionOperatorState implements SinkFunction<Tuple2<String, Integer>>, CheckpointedFunction {

        private ListState<Tuple2<String, Integer>> checkpointState;


        private List<Tuple2<String, Integer>> elements;
        private int thershold;

        public SinkFunctionOperatorState(int thershold) {
            this.thershold = thershold;
            this.elements = new ArrayList();
        }

        @Override
        public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
            elements.add(value);

            if (elements.size() == thershold) {
                System.out.println("数据：\t" + elements);
                elements.clear();
            }

        }

        /**
         * ÎΩ
         * 定时存快照
         *
         * @param functionSnapshotContext
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            checkpointState.clear();
            elements.forEach(x -> {
                try {
                    checkpointState.add(x);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }

        /**
         * 程序启动的时候
         *
         * @param functionInitializationContext
         * @throws Exception
         */
        @Override
        public void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {

            ListStateDescriptor<Tuple2<String, Integer>> descriptor = new ListStateDescriptor<>(
                    "buffer",
                    TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
                    })
            );

            checkpointState = functionInitializationContext.getOperatorStateStore().getListState(descriptor);
            if (functionInitializationContext.isRestored()) {
                checkpointState.get().iterator().forEachRemaining(x -> elements.add(x));
            }
        }
    }


}
